Reactive @KafkaListener [DO NOT MERGE] #1123

garyrussell · 2019-06-13T20:14:31Z

Initial PoC only.

artembilan

Just a couple concerns from me.

samples/sample-04/kafka-4-reactor/src/main/java/com/example/Kafka4ReactorApplication.java

artembilan · 2019-06-14T15:45:27Z

samples/sample-04/kafka-4-reactor/src/main/java/com/example/Kafka4ReactorApplication.java

+class Listener {
+
+	@KafkaListener(topics = "skReactorTopic")
+	public Disposable listen(Flux<ReceiverRecord<String, String>> flux) {


I think the return type must be Mono<Void> to postpone a subscription to the container.
Similar way what we have with WebFlux and RSocket implementations in Spring.

Yes, I agree with Artem that it would be better to return Mono<Void> here

artembilan · 2019-06-14T15:46:35Z

@bsideup,

We would appreciate to have your feedback on this one since you are familiar with Reactor Kafka.

Thank you in advance!

bsideup · 2019-06-14T15:58:02Z

spring-kafka/src/main/java/org/springframework/kafka/listener/reactive/ReactorAdapter.java

+
+	@Override
+	public void stop() {
+		if (this.disposable != null && !this.disposable.isDisposed()) {


no need to check for isDisposed, it should be idempotent

hint: you can pre-initialize the field with something like static final Disposable DISPOSED = Disposable.disposed() to avoid the null checks

…he registry.

artembilan

A couple concerns so far.

artembilan · 2019-06-20T17:31:11Z

...main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

 import org.springframework.util.StringUtils;
 import org.springframework.validation.Validator;

+import reactor.core.publisher.Flux;


With this Reactor becomes as not a optional dependency any more...
Is it really what we are aiming now ?

artembilan · 2019-06-20T17:35:18Z

spring-kafka/src/main/java/org/springframework/kafka/listener/reactive/ReactorAdapter.java

+	public void start() {
+		ReceiverOptions<?, ?> options = ReceiverOptions.create(this.configs)
+				.subscription(this.topics);
+		Flux<?> flux = KafkaReceiver.create(options).receive();


Anything wrong with using ReactiveKafkaConsumerTemplate instead?

garyrussell · 2019-08-14T13:18:21Z

Closing for now.

hartmut-co-uk · 2020-02-04T00:49:05Z

Hi, has there been issues with the implementation so far - or has it simply been de-scoped for now?

garyrussell · 2020-02-04T13:39:35Z

We just haven't had the bandwidth to work on it; we have also discovered some problems with the reactor-kafka project with respect to producer fencing when transactions are being used. That issue needs to be addressed over there; I hope to find some time to look at it soon.

garyrussell · 2020-11-23T18:44:09Z

Closing for now.

roger751 · 2021-12-15T18:46:36Z

@garyrussell can you update the progress about this subject? I have some questions which I can't find answers to:

What were the problems with Reactor-Kafka and transactions?
Have they been solved?
Is there any alternative to this subject or does it still in plans for being implemented?

garyrussell · 2021-12-15T18:55:05Z

The problem is around producer fencing - when using EOS mode ALPHA (now called V1), the transactional id needs to be based on the group/topic/partition; currently it processes the entire ConsumerRecords in a single transaction and the records might be from different partitions.
It is solved by using EOS mode BETA (now called V2), but that needs some work Add Support for EOS Mode BETA (aka V2) reactor/reactor-kafka#207
As I said on the other issue in reactor-kafka - you need to explain exactly what use case you are trying to satisfy. There is an impedance mismatch between the projects so it's hard to come up with a general solution for all possible use cases.

roger751 · 2021-12-15T19:38:43Z

I understand. Thank you!

roger751 · 2022-01-03T23:07:15Z

Hi @garyrussell, I'm reading this thread again and I wonder if after the PR will be merged reactor/reactor-kafka#258 you plan of reactivating this current PR for a reactive @KafkaListener? As for your question about my use case, I think it's the most general and common one - I'm trying to read from Kafka in a Reactor based application, and I'm willing to use the most out of the great Spring Boot and Cloud features. Ideally, I would be able to do that by using Reactor Kafka with Spring Cloud Stream. A reactive @KafkaListener is just a code reducer and a much more comfortable way of writing code in my opinion.

garyrussell · 2022-01-03T23:17:09Z

I don't see any "code reducing" here. Reactor Kafka is doing the same work as the listener container; the goal being to get records from the broker and pass them to user code. The listener container and consumer event loop are effectively doing the same thing.

KafkaReceiver.create(ro)
	.receive()
	.publishOn(Schedulers.single(), 1)
	.doOnNext(record -> {
            processIt(record);
	})
        ...
        .subscribe();

I am struggling to see the value add; a concrete example might help.

harishvashistha · 2023-01-06T16:16:03Z

@garyrussell the above code work perfect as you suggested and working fine for me. But can you help with the example where reactor-kafka can concurrently subscribe from multiple partitions of the topic concurrently in the same service instance. I don't want to leave reactor kafka just in absence of this feature. May be I am not aware. Please help.

garyrussell · 2023-01-09T16:07:04Z

@harishvashistha It is not clear what you mean; you can subscribe to multiple topics in the ReceiverOptions.subscription(). Kafka will assign as many partitions as possible to each consumer (with each consumer in the same group each getting different partitions). If there is only one consumer, it will get all partitions.

harishvashistha · 2023-01-09T16:59:06Z

@garyrussell I mean, I have this code which returns ServerSentEvent to SSE API over HTTP by fetching records from kafka. But this creates a single consumer to kafka topic with 20 partition:

public Flux<ServerSentEvent<String>> getServerSentEvents() {
        final KafkaReceiver<Key, ValueSse> kafkaReceiverFromLastOffset =  kafkaReceiverProvider.getObject();

        return kafkaReceiver.receive()
                .publishOn(Schedulers.fromExecutor(taskExecutor))
                .log()
                .doOnEach(receiverRecordSignal -> log.trace("message received, from queue: {} ", receiverRecordSignal.get()))
                .map(ConsumerRecord::value)
                .map(valueSse -> ServerSentEvent.builder().data(valueSse).build())
                .doOnError(e -> log.error("invalid payload received while processing kafka message ", e))
                .onErrorContinue((e, record) -> {
                    log.error(MESSAGE_PARSING_FAILED, e);
                    Flux.empty();
                });
    }

@Bean
@Scope("prototype")
public KafkaReceiver<Key, Value> kafkaReceiver() throws ExecutionException, InterruptedException {
        final String consumerId = UUID.randomUUID().toString();
        final Map<String, Object> propsMaps = kafkaReceiverConfigurations(consumerId);

        return new DefaultKafkaReceiver(ConsumerFactory.INSTANCE,
                ReceiverOptions.create(propsMaps)
                        .addAssignListener(receiverPartitions -> {
                            for (final ReceiverPartition p : receiverPartitions) {
                                p.seekToEnd();
                            }
                        })
                        .assignment(getTopicPartitions()));
    }

As you said above, this kafkaReceiver.receive() just creates a single kafka consumer for all the partitions of the kafka topic.

But how can I create multiple kafka consumers reading data from multiple partitions parallely as per my requirement. As I have to return this Flux of ServerSentEvent to the caller client over SSE API?

I have followed this thread as well shared in https://stackoverflow.com/questions/69891782/multi-threading-on-kafka-send-in-spring-reactor-kafka . But not sure how to use this code to return Flux of ServerSentEvent to calling API:

@Bean
    public ReceiverOptions<String, String> kafkaReceiverOptions(String topic, KafkaProperties kafkaProperties) {
        ReceiverOptions<String, String> basicReceiverOptions = ReceiverOptions.create(kafkaProperties.buildConsumerProperties());
        return basicReceiverOptions.subscription(Collections.singletonList(topic))
                .addAssignListener(receiverPartitions -> log.debug("onPartitionAssigned {}", receiverPartitions))
                .addRevokeListener(receiverPartitions -> log.debug("onPartitionsRevoked {}", receiverPartitions));
    }

@Bean
public ReactiveKafkaConsumerTemplate<String, String> kafkaConsumerTemplate(ReceiverOptions<String, String> kafkaReceiverOptions) {
    return new ReactiveKafkaConsumerTemplate<String, String>(kafkaReceiverOptions);
}

public void run(String... args) {
        for(int i = 0; i < topicPartitionsCount ; i++) {
            readWrite(destinationTopic).subscribe();
        }
    }}

public Flux<String> readWrite(String destTopic) {
        return kafkaConsumerTemplate
                .receiveAutoAck()
                .doOnNext(consumerRecord -> log.info("received key={}, value={} from topic={}, offset={}",
                        consumerRecord.key(),
                        consumerRecord.value(),
                        consumerRecord.topic(),
                        consumerRecord.offset())
                )
                .map(ConsumerRecord::value)               
                .onErrorContinue((exception,errorConsumer)->{
                    log.error("Error while consuming : {}", exception.getMessage());
                });
    }

garyrussell · 2023-01-09T17:04:59Z

You would need to create multiple receivers and merge their results into a single flux.

harishvashistha · 2023-01-09T18:08:04Z

@garyrussell are you hinting that something like this would be required?:

 public Flux<ServerSentEvent<String>> getSSEDataFromMultiplePartitions() {
        Flux<ServerSentEvent<String>> toReturn = Flux.empty();
        for (int i = 0; i < partitionCount; i++) {
            Flux<ServerSentEvent<String>> eventsFromDedicatedConnection = getServerSentEvents();
            toReturn.mergeWith(eventsFromDedicatedConnection);
        }
        return toReturn;
    }

As this is not returning any data to SSE channel at all.

public Flux<ServerSentEvent<String>> getServerSentEvents() {
        final KafkaReceiver<Key, ValueSse> kafkaReceiverFromLastOffset =  kafkaReceiverProvider.getObject();

        return kafkaReceiver.receive()
                .publishOn(Schedulers.fromExecutor(taskExecutor))
                .log()
                .doOnEach(receiverRecordSignal -> log.trace("message received, from queue: {} ", receiverRecordSignal.get()))
                .map(ConsumerRecord::value)
                .map(valueSse -> ServerSentEvent.builder().data(valueSse).build())
                .doOnError(e -> log.error("invalid payload received while processing kafka message ", e))
                .onErrorContinue((e, record) -> {
                    log.error(MESSAGE_PARSING_FAILED, e);
                    Flux.empty();
                });
    }

@Bean
@Scope("prototype")
public KafkaReceiver<Key, Value> kafkaReceiver() throws ExecutionException, InterruptedException {
        final String consumerId = UUID.randomUUID().toString();
        final Map<String, Object> propsMaps = kafkaReceiverConfigurations(consumerId);

        return new DefaultKafkaReceiver(ConsumerFactory.INSTANCE,
                ReceiverOptions.create(propsMaps)
                        .addAssignListener(receiverPartitions -> {
                            for (final ReceiverPartition p : receiverPartitions) {
                                p.seekToEnd();
                            }
                        })
                        .assignment(getTopicPartitions()));
    }

garyrussell · 2023-01-09T18:14:49Z

Sorry; I am not a reactor expert; you'd be better off asking somewhere like stack overflow.

That said, you seem to be discarding the new flux created by mergeWith()...

toReturn.mergeWith(eventsFromDedicatedConnection);

Looks like it should be

toReturn = toReturn.mergeWith(eventsFromDedicatedConnection);

artembilan requested changes Jun 14, 2019

View reviewed changes

bsideup reviewed Jun 14, 2019

View reviewed changes

garyrussell added 3 commits June 17, 2019 12:08

Reactive @KafkaListener

7d6a87f

Polish comment

833e1e4

PR Comments; return Mono<Void> instead of Disposable.

2649b72

garyrussell force-pushed the reactorListener branch from acddf0c to 2649b72 Compare June 17, 2019 16:23

garyrussell mentioned this pull request Jun 17, 2019

How to implement kafka listeners for the reactor kafka consumer reactor/reactor-kafka#100

Closed

Support concurrency, group.id, client.id and create the adapters in t…

ca26e06

…he registry.

artembilan requested changes Jun 20, 2019

View reviewed changes

garyrussell added this to the Backlog milestone Aug 14, 2019

hartmut-co-uk mentioned this pull request Feb 4, 2020

Sample needed | Long running Kafka consumer reactor/reactor-kafka#62

Closed

garyrussell closed this Nov 23, 2020

garyrussell removed this from the Backlog milestone Oct 20, 2022

Reactive @KafkaListener [DO NOT MERGE] #1123

Reactive @KafkaListener [DO NOT MERGE] #1123

Uh oh!

Conversation

garyrussell commented Jun 13, 2019

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

artembilan Jun 14, 2019

Choose a reason for hiding this comment

Uh oh!

bsideup Jun 14, 2019

Choose a reason for hiding this comment

Uh oh!

artembilan commented Jun 14, 2019

Uh oh!

bsideup Jun 14, 2019

Choose a reason for hiding this comment

Uh oh!

artembilan left a comment

Choose a reason for hiding this comment

Uh oh!

artembilan Jun 20, 2019

Choose a reason for hiding this comment

Uh oh!

artembilan Jun 20, 2019

Choose a reason for hiding this comment

Uh oh!

garyrussell commented Aug 14, 2019

Uh oh!

hartmut-co-uk commented Feb 4, 2020

Uh oh!

garyrussell commented Feb 4, 2020

Uh oh!

garyrussell commented Nov 23, 2020

Uh oh!

roger751 commented Dec 15, 2021

Uh oh!

garyrussell commented Dec 15, 2021

Uh oh!

roger751 commented Dec 15, 2021

Uh oh!

roger751 commented Jan 3, 2022

Uh oh!

garyrussell commented Jan 3, 2022

Uh oh!

harishvashistha commented Jan 6, 2023

Uh oh!

garyrussell commented Jan 9, 2023

Uh oh!

harishvashistha commented Jan 9, 2023

Uh oh!

garyrussell commented Jan 9, 2023

Uh oh!

harishvashistha commented Jan 9, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

garyrussell commented Jan 9, 2023

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

harishvashistha commented Jan 9, 2023 •

edited

Loading